package zookeeper.lock;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.retry.RetryNTimes;
import org.apache.curator.retry.RetryUntilElapsed;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @author Arnold Yand
 * @summary Apache Curator
 * @time 2018/4/15 17:58
 */
public class Curator {
  private static final String rootPath = "/etl_curator";
  private static final RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
  private Integer data = 0;


  public static void main(String[] args) throws Exception {
    new Curator().accumulateAdd();
    Thread.yield();
//    TimeUnit.SECONDS.sleep(60);
  }

  public void accumulateAdd() throws Exception {
    CuratorFramework client1 = CuratorFrameworkFactory.builder()
            .connectString("127.0.0.1:2181")
            .sessionTimeoutMs(50000)//会话超时时间
            .connectionTimeoutMs(50000)//连接超时时间
            .retryPolicy(retryPolicy)
            .build();
    client1.start();
    client1.delete().deletingChildrenIfNeeded().forPath(rootPath);
    client1.create().creatingParentContainersIfNeeded()
            .withMode(CreateMode.PERSISTENT)
            .forPath(rootPath,data.toString().getBytes());

    int m = 10;
    final CountDownLatch countDownLatch = new CountDownLatch(m);
    int n = 2;
    for(int i = 0;i<m;i++){
      new Thread(new Runnable() {
        @Override
        public void run() {
          CuratorFramework client = CuratorFrameworkFactory.builder()
                  .connectString("127.0.0.1:2181")
                  .sessionTimeoutMs(50000)//会话超时时间
                  .connectionTimeoutMs(50000)//连接超时时间
                  .retryPolicy(retryPolicy)
                  .build();
          client.start();

          int count = 0;
          Integer value = null;
          while (true) {
            try{
              Stat stat = new Stat();
              // 获取当前value和状态信息
              value = Integer.valueOf(new String(client.getData().storingStatIn(stat).forPath(rootPath)));
              value = value + 1;
              client.setData().withVersion(stat.getVersion()).forPath(rootPath, value.toString().getBytes());
              count++;
              System.out.println(Thread.currentThread().getName() + "--- count:" + count +", value="+value);
              if(count == n){
                break;
              }
            }catch (KeeperException.BadVersionException bve){
              // dont worries
              bve.printStackTrace();
            } catch (Exception e) {
              e.printStackTrace();
              break;
            }
          }
          countDownLatch.countDown();
        }
      }).start();
      System.out.println("thread "+i);
    }

    countDownLatch.await();
  }

  public void curatorTest() throws Exception {

//    RetryPolicy retryPolicy1 = new RetryNTimes(3, 1000);//最大重试次数，和两次重试间隔时间
//    RetryPolicy retryPolicy2 = new RetryUntilElapsed(5000, 1000);//会一直重试直到达到规定时间，第一个参数整个重试不能超过时间，第二个参数重试间隔
//    //第一种方式
//    CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.0.3:2181", 5000, 5000, retryPolicy);//最后一个参数重试策略

    CuratorFramework client = CuratorFrameworkFactory.builder()
            .connectString("127.0.0.1:2181")
            .sessionTimeoutMs(5000)//会话超时时间
            .connectionTimeoutMs(5000)//连接超时时间
            .retryPolicy(retryPolicy)
            .build();

    client.start();

    String rootPath = "/etl_curator";
    String path = client.create().creatingParentsIfNeeded()//若创建节点的父节点不存在会先创建父节点再创建子节点
            .withMode(CreateMode.PERSISTENT)
            .forPath(rootPath,"158".getBytes());
    System.out.println("path:"+path);

    List<String> list = client.getChildren().forPath(rootPath);
    System.out.println(list);

    //String re = new String(client.getData().forPath("/curator/3"));//只获取数据内容
    Stat stat = new Stat();
    //在获取节点内容的同时把状态信息存入Stat对象
    String re = new String(client.getData().storingStatIn(stat).forPath(rootPath));
    System.out.println(re);
    System.out.println(stat);

    client.getData().storingStatIn(stat);
    client.setData().withVersion(stat.getVersion())//修改前获取一次节点数据得到版本信息
            .forPath(rootPath, "111".getBytes());

    client.delete().guaranteed()//保障机制，若未删除成功，只要会话有效会在后台一直尝试删除
            .deletingChildrenIfNeeded()//若当前节点包含子节点
            .withVersion(-1)//指定版本号
            .forPath(rootPath);
    TimeUnit.SECONDS.sleep(60);
  }
}
